package cn.doitedu.dw.util

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.types.{DataTypes, StructType}

object IdBindRollCombine {
  def main(args: Array[String]): Unit = {

    System.setProperty("HADOOP_USER_NAME", "atguigu")
    Logger.getLogger("org").setLevel(Level.WARN)

    val spark = SparkSession.builder()
      .enableHiveSupport()
      .config("spark.sql.shuffle.partitions", "2")
      .appName("地理位置知识库加工")
      .master("local")
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._

    val schema = new StructType()
      .add("account",DataTypes.StringType)
      .add("deviceid", DataTypes.StringType)
      .add("sessionid", DataTypes.StringType)
      .add("ts", DataTypes.LongType)

    val logDf = spark.read.table("ods17.app_action_log").where(s"dt='2020-10-07'")
    logDf.createTempView("logdf")

    val loginCnts: DataFrame = spark.sql(
      """
        |select
        |
        |deviceid,
        |account,
        |min(timestamp) as first_login_ts,
        |count(distinct sessionid)*100 as bind_score
        |from logdf
        |group by deviceid,account
        |""".stripMargin)
    loginCnts.createTempView("today")

    println("当天评分结果")
    loginCnts.show(100,false)

    val bindScorePre: DataFrame = spark.read.table("dwd17.id_account_bind").where("dt='2020-10-06'")

    println("历史评分结果")
    bindScorePre.show(100,false)
    bindScorePre.createTempView("yestoday")

    val combined = spark.sql(
      """
        | insert into table dwd17.id_account_bind partition(dt='2020-10-07')
        |
        |select
        |
        |if(today.deviceid is null,yestoday.deviceid,today.deviceid) as deviceid,
        |if(today.account is null,yestoday.account,today.account) as account,
        |if(yestoday.first_login_ts is not null,yestoday.first_login_ts,today.first_login_ts) as first_login_ts,
        |-- if(today.account is null,yestoday.login_cnt,today.login_cnt+yestoday.login_cnt) as login_cnt,
        |if(today.account is null,yestoday.bind_score*0.9,today.bind_score+if(yestoday.bind_score is null,0,yestoday.bind_score)) as bind_score
        |from
        |  today
        |full join
        |  yestoday
        |on today.deviceid=yestoday.deviceid and today.account=yestoday.account
        |
        |""".stripMargin)

    combined.show(100)

    spark.close()
  }

}
